package com.cmiov.pulsar.pulsardemo;

import org.apache.pulsar.functions.api.Context;
import org.apache.pulsar.functions.api.Function;

import java.util.Arrays;

/**
 * * Licensed to CMDT,Inc. under the terms of the CMDT
 * Software License version 1.0.
 * <p>
 * See the NOTICE file distributed with this work for additional
 * information regarding copyright ownership.
 * ----------------------------------------------------------------------------
 * Date           Author               Version        Comments
 * 18-12-24       joe chow                 1.0            Initial Version
 */
public class WordCountFunction implements Function<String, Integer> {
    @Override
    public Integer process(String s, Context context) throws Exception {
        Arrays.asList(s.split("\\s+")).forEach(word -> {
            String countKey = word.toLowerCase();
            context.incrCounter(countKey, 1);
            context.recordMetric(countKey, 1);
        });

        int length = s.split("\\s+").length;
        if (length > 5) {
            context.getLogger().warn("{} is too long! length = {}", s, length);
        } else {
            context.getLogger().info("{}'s length is {}", s, length);
        }
        return length;
    }
}
